package com.pisolution.scheduler.runtime.master.scheduler;

import com.pisolution.scheduler.core.utils.OSUtils;
import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.datasource.command.Command;
import com.pisolution.scheduler.datasource.instance.WorkflowInstance;
import com.pisolution.scheduler.runtime.master.MasterConfig;
import com.pisolution.scheduler.runtime.process.ProcessService;
import com.pisolution.scheduler.runtime.registry.FaultTolerant;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.Closeable;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 管理节点调度服务.
 *
 * @author sora
 * @version 0.1.0
 */
@Service
public class MasterSchedulerService extends Thread implements Closeable {
    private static final Logger LOGGER = LoggerFactory.getLogger(MasterSchedulerService.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于读取调度服务相关配置. */
    private final MasterConfig config;

    /** 用于处理系统控制指令. */
    private final ProcessService service;

    /** 用于监听注册中心连接状态. */
    private final FaultTolerant tolerant;

    // -------------------------------------------------------------------------------------------------

    /** 用于确认执行指令所需的线程数是否充足. */
    private ThreadPoolExecutor executor;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化调度服务.
     *
     * @param config 用于读取调度服务相关配置.
     * @param service 用于处理系统控制指令.
     * @param tolerant 用于监听注册中心连接状态.
     */
    public MasterSchedulerService(final MasterConfig config, final ProcessService service, final FaultTolerant tolerant) {
        this.config = config;
        this.service = service;
        this.tolerant = tolerant;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化线程环境.
     *
     * <p><b>注意:</b></p>
     * <p>该方法会在构造函数与依赖注入之后自动调用 无需手动触发.</p>
     */
    @SuppressWarnings("unused")
    @PostConstruct
    public void init() {
        this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(
            this.config.getMasterSchedulerThreads(),
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("master-scheduler-thread").build()
        );
    }

    @Override
    public synchronized void start() {
        super.setName("master-scheduler");
        super.start();
    }

    @SuppressWarnings("BusyWait")
    @Override
    public void run() {
        LOGGER.info("The master scheduler service started.");

        while (Stopper.isRunning()) {
            InterProcessMutex mutex = null;

            try {
                // 1. 校验当前可用资源是否充足.
                final double loadAverage = OSUtils.loadAverage();
                final double availablePhysicalMemory = OSUtils.availablePhysicalMemory();

                if (loadAverage > this.config.getMaxCPULoadAvg() || availablePhysicalMemory < this.config.getReservedMemory()) {
                    LOGGER.warn("The processor load is too high or the available physical memory is too low.");
                    Thread.sleep(1000);

                    continue;
                }

                // 2. 校验注册中心是否已经启动.
                if (this.tolerant.getClient().getState() != CuratorFrameworkState.STARTED) {
                    continue;
                }

                // 3. 锁住特定路径 (/lock/masters) 以进行调度.
                mutex = new InterProcessMutex(this.tolerant.getClient(), "/lock/masters");
                mutex.acquire();

                // 4. 查找下一个需要调度的系统指令.
                final Command command = this.service.findOneCommand();

                // 5. 确认是否有系统指令需要进行调度.
                if (command == null) {
                    Thread.sleep(1000);

                    continue;
                }

                LOGGER.info("Find one command {} of type {}.", command.getId(), command.getType());

                try {
                    // 6. 从指令中解析出相应的工作流实例对象以完成指令.
                    final WorkflowInstance instance = this.service.handleCommand(
                        LOGGER,
                        OSUtils.getHost() + ":" + this.config.getPort(),
                        this.config.getMasterSchedulerThreads() - this.executor.getActiveCount(),
                        command
                    );
                    if (instance != null) {
                        LOGGER.info("Start the master scheduler thread and split the directed acyclic graph.");

                        // 7. 执行解析出来的工作流实例.
                        this.executor.execute(new WorkflowScheduler(instance, this.service));
                    }
                } catch (Exception e) {
                    // 8. 处理失败 将该指令标记为运行失败的指令并从待运行的指令集集合中移除该指令.
                    LOGGER.error("An error occurred while scanning the command.", e);

                    this.service.moveToErrorCommand(command, e.toString());
                }
            } catch (Exception e) {
                LOGGER.error("An error occurred while scheduling the command.", e);
            } finally {
                this.tolerant.releaseMutex(mutex);
            }
        }
    }

    @Override
    public void close() {
        this.executor.shutdown();

        try {
            if (!this.executor.awaitTermination(5, TimeUnit.SECONDS)) {
                LOGGER.warn("The master scheduler service shutdown without terminated.");
            }
        } catch (InterruptedException ignore) {
            // 无需处理此异常.
        }

        LOGGER.info("The master scheduler service stopped.");
    }
}
